// Tencent is pleased to support the open source community by making
// 蓝鲸智云 - 监控平台/日志平台 (BlueKing - Monitor/Log) available.
// Copyright (C) 2017-2022 THL A29 Limited, a Tencent company. All rights reserved.
// Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
// You may obtain a copy of the License at http://opensource.org/licenses/MIT
// Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
// an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
// specific language governing permissions and limitations under the License.
//

package otlp

import (
	"context"
	"time"

	"github.com/TencentBlueKing/bkmonitor-kits/logger"
	"go.opentelemetry.io/collector/model/otlpgrpc"
	"google.golang.org/grpc"
	"google.golang.org/grpc/peer"

	"github.com/TencentBlueKing/bk-collector/define"
	"github.com/TencentBlueKing/bk-collector/internal/monitoring"
	"github.com/TencentBlueKing/bk-collector/receiver"
)

func init() {
	receiver.RegisterGrpcRoute(func(s *grpc.Server) {
		otlpgrpc.RegisterTracesServer(s, grpcsrv.traces)
		otlpgrpc.RegisterMetricsServer(s, grpcsrv.metrics)
		otlpgrpc.RegisterLogsServer(s, grpcsrv.logs)
	})
}

type grpcSrv struct {
	traces  otlpgrpc.TracesServer
	metrics otlpgrpc.MetricsServer
	logs    otlpgrpc.LogsServer
}

var grpcsrv = grpcSrv{
	traces:  tracesSrv{},
	metrics: metricsSrv{},
	logs:    logsSrv{},
}

func getIpFromContext(ctx context.Context) string {
	if p, ok := peer.FromContext(ctx); ok {
		return define.ParseRequestIP(p.Addr.String())
	}
	return ""
}

type tracesSrv struct{}

func (tracesSrv) Export(ctx context.Context, req otlpgrpc.TracesRequest) (otlpgrpc.TracesResponse, error) {
	lvs := []string{Source, string(define.RequestGrpc), string(define.RecordTraces)}

	traces := req.Traces()
	numSpans := traces.SpanCount()
	if numSpans == 0 {
		monitoring.ReceiverSkippedTotal.WithLabelValues(lvs...).Inc()
		return otlpgrpc.NewTracesResponse(), nil
	}

	start := time.Now()
	receiver.PublishRecord(&define.Record{
		RequestType:   define.RequestGrpc,
		RequestClient: define.RequestClient{IP: getIpFromContext(ctx)},
		RecordType:    define.RecordTraces,
		Data:          traces,
	})

	for i := 0; i < traces.ResourceSpans().Len(); i++ {
		instSpans := traces.ResourceSpans().At(i).InstrumentationLibrarySpans()
		for j := 0; j < instSpans.Len(); j++ {
			spans := instSpans.At(j).Spans()
			for k := 0; k < spans.Len(); k++ {
				span := spans.At(k)
				logger.Debugf("traceId=%v, spanId=%v", span.TraceID().HexString(), span.SpanID().HexString())
			}
		}
	}

	monitoring.ReceiverReceivedTotal.WithLabelValues(lvs...).Add(float64(traces.SpanCount()))
	monitoring.ReceiverHandledDuration.WithLabelValues(lvs...).Observe(float64(time.Since(start).Milliseconds()))
	monitoring.ReceiverHandledTotal.WithLabelValues(lvs...).Inc()
	return otlpgrpc.NewTracesResponse(), nil
}

type metricsSrv struct{}

func (metricsSrv) Export(ctx context.Context, req otlpgrpc.MetricsRequest) (otlpgrpc.MetricsResponse, error) {
	lvs := []string{Source, string(define.RequestGrpc), string(define.RecordMetrics)}
	metrics := req.Metrics()
	dataPointCount := metrics.DataPointCount()
	if dataPointCount == 0 {
		monitoring.ReceiverSkippedTotal.WithLabelValues(lvs...).Inc()
		return otlpgrpc.NewMetricsResponse(), nil
	}

	start := time.Now()
	receiver.PublishRecord(&define.Record{
		RequestType:   define.RequestGrpc,
		RequestClient: define.RequestClient{IP: getIpFromContext(ctx)},
		RecordType:    define.RecordMetrics,
		Data:          metrics,
	})
	monitoring.ReceiverReceivedTotal.WithLabelValues(lvs...).Add(float64(metrics.MetricCount()))
	monitoring.ReceiverHandledDuration.WithLabelValues(lvs...).Observe(float64(time.Since(start).Milliseconds()))
	monitoring.ReceiverHandledTotal.WithLabelValues(lvs...).Inc()
	return otlpgrpc.NewMetricsResponse(), nil
}

type logsSrv struct{}

func (logsSrv) Export(ctx context.Context, req otlpgrpc.LogsRequest) (otlpgrpc.LogsResponse, error) {
	lvs := []string{Source, string(define.RequestGrpc), string(define.RecordLogs)}
	logs := req.Logs()
	numSpans := logs.LogRecordCount()
	if numSpans == 0 {
		monitoring.ReceiverSkippedTotal.WithLabelValues(lvs...).Inc()
		return otlpgrpc.NewLogsResponse(), nil
	}

	start := time.Now()
	receiver.PublishRecord(&define.Record{
		RequestType:   define.RequestGrpc,
		RequestClient: define.RequestClient{IP: getIpFromContext(ctx)},
		RecordType:    define.RecordLogs,
		Data:          logs,
	})
	monitoring.ReceiverReceivedTotal.WithLabelValues(lvs...).Add(float64(logs.LogRecordCount()))
	monitoring.ReceiverHandledDuration.WithLabelValues(lvs...).Observe(float64(time.Since(start).Milliseconds()))
	monitoring.ReceiverHandledTotal.WithLabelValues(lvs...).Inc()
	return otlpgrpc.NewLogsResponse(), nil
}
